/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.dcloud.core.client.rdbms.absRdbms;


import com.dcloud.core.exception.DtLoaderException;
import com.dcloud.core.source.ISource;
import com.dcloud.core.source.RdbmsSource;
import com.dcloud.core.util.DBUtil;
import com.dcloud.core.util.PropertiesUtil;
import com.dcloud.core.util.ReflectUtil;
import com.zaxxer.hikari.HikariDataSource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;
import java.util.Properties;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @Description //关系型数据库连接工厂
 * @Author shijf
 * @Date 0:17 2022/7/7
 */
@Slf4j
public  class AbsRdbmsConnFactory {
    protected String driverName = null;

    protected String testSql;

    private static ConcurrentHashMap<String, Object> hikariDataSources = new ConcurrentHashMap<>();

    private AtomicBoolean isFirstLoaded = new AtomicBoolean(true);

    private static final String CP_POOL_KEY = "url:%s,username:%s,password:%s,properties:%s";

    /**
     * 线程池 - 用于部分数据源获取连接超时处理
     */
    //protected static ExecutorService executor = new ThreadPoolExecutor(5, 10, 1L, TimeUnit.MINUTES, new LinkedBlockingQueue<>(1000));

    protected void init() throws ClassNotFoundException {
        // 减少加锁开销
        if (!isFirstLoaded.get()) {
            return;
        }

        synchronized (AbsRdbmsConnFactory.class) {
            if (isFirstLoaded.get()) {
                Class.forName(driverName);
                isFirstLoaded.set(false);
            }
        }
    }

    /**
     * 获取连接，对抛出异常进行统一处理
     *
     * @param iSource
     * @return
     * @throws Exception
     */
    @Deprecated
    public Connection getConn(ISource iSource) throws Exception {
        return getConn(iSource, StringUtils.EMPTY);
    }

    /**
     * 获取连接，对抛出异常进行统一处理
     *
     * @param iSource
     * @param taskParams
     * @return
     * @throws Exception
     */
    public Connection getConn(ISource iSource, String taskParams) throws Exception {
        if (iSource == null) {

        }
        try {
            RdbmsSource rdbmsSource = (RdbmsSource) iSource;
            boolean isStart = rdbmsSource.getPoolConfig() != null;
            Connection connection = isStart && MapUtils.isEmpty(rdbmsSource.getKerberosConfig()) ?
                    getCpConn(rdbmsSource) : getSimpleConn(rdbmsSource);

            return setSchema(connection, rdbmsSource.getSchema());
        } catch (Exception e){
            e.printStackTrace();
            // 对异常进行统一处理
            DtLoaderException dtLoaderException = new DtLoaderException(e.getMessage());
            String message = dtLoaderException.getMessage();
            throw new DtLoaderException(e.getMessage(),e);
        }
    }

    /**
     * 从连接池获取连接
     *
     * @param source
     * @return
     * @throws Exception
     */
    protected Connection getCpConn(ISource source) throws Exception {
        RdbmsSource RdbmsSource = (RdbmsSource) source;
        String poolKey = getPrimaryKey(RdbmsSource);
        log.info("datasource connected(Hikari), url : {}, userName : {}, kerberosConfig : {}", RdbmsSource.getUrl(), RdbmsSource.getUsername(), RdbmsSource.getKerberosConfig());
        HikariDataSource hikariData = (HikariDataSource) hikariDataSources.get(poolKey);
        if (hikariData == null) {
            synchronized (AbsRdbmsConnFactory.class) {
                hikariData = (HikariDataSource) hikariDataSources.get(poolKey);
                if (hikariData == null) {
                    hikariData = transHikari(source);
                    hikariDataSources.put(poolKey, hikariData);
                    return hikariData.getConnection();
                }
            }
        }

        return hikariData.getConnection();
    }

    /**
     * 获取普通连接
     *
     * @param source
     * @return
     * @throws Exception
     */
    protected Connection getSimpleConn(ISource source) throws Exception {
        RdbmsSource RdbmsSource = (RdbmsSource) source;
        init();
        DriverManager.setLoginTimeout(30);
        String url = dealSourceUrl(RdbmsSource);
        log.info("datasource connected, url : {}, userName : {}, kerberosConfig : {}", url, RdbmsSource.getUsername(), RdbmsSource.getKerberosConfig());
        return DriverManager.getConnection(url, PropertiesUtil.convertToProp(RdbmsSource));
    }

    /**
     * 处理 URL 地址
     *
     * @param RdbmsSource
     * @return
     */
    protected String dealSourceUrl(RdbmsSource RdbmsSource) {
        return RdbmsSource.getUrl();
    }

    public Boolean testConn(ISource source) {
        Connection conn = null;
        Statement statement = null;
        try {
            conn = getConn(source, StringUtils.EMPTY);
            if (StringUtils.isBlank(testSql)) {
                conn.isValid(5);
            } else {
                statement = conn.createStatement();
                statement.execute((testSql));
            }
            return true;
        } catch (Exception e) {
            throw new DtLoaderException(e.getMessage(), e);
        }finally {
            DBUtil.closeDBResources(null, statement, conn);
        }
    }

    /**
     * sourceDTO 转化为 HikariDataSource
     *
     * @param source
     * @return
     */
    protected HikariDataSource transHikari(ISource source) {
        RdbmsSource RdbmsSource = (RdbmsSource) source;
        HikariDataSource hikariData = new HikariDataSource();

        // 设置 driverClassName
        String driverClassName = getDriverClassName(source);
        hikariData.setDriverClassName(driverClassName);
        hikariData.setUsername(RdbmsSource.getUsername());
        hikariData.setPassword(RdbmsSource.getPassword());
        hikariData.setJdbcUrl(RdbmsSource.getUrl());
        hikariData.setConnectionInitSql(testSql);

        hikariData.setConnectionTestQuery(testSql);
        hikariData.setConnectionTimeout(RdbmsSource.getPoolConfig().getConnectionTimeout());
        hikariData.setIdleTimeout(RdbmsSource.getPoolConfig().getIdleTimeout());
        hikariData.setMaxLifetime(RdbmsSource.getPoolConfig().getMaxLifetime());
        hikariData.setMaximumPoolSize(RdbmsSource.getPoolConfig().getMaximumPoolSize());
        hikariData.setMinimumIdle(RdbmsSource.getPoolConfig().getMinimumIdle());
        hikariData.setReadOnly(RdbmsSource.getPoolConfig().getReadOnly());

        Properties properties = PropertiesUtil.convertToProp(RdbmsSource);
        for (Object key : properties.keySet()) {
            hikariData.addDataSourceProperty(key.toString(), properties.get(key));
        }
        return hikariData;
    }

    protected String getDriverClassName(ISource source) {
        return driverName;
    }

    /**
     * 根据数据源获取数据源唯一 KEY
     *
     * @param source
     * @return
     */
    protected String getPrimaryKey(ISource source) {
        RdbmsSource RdbmsSource = (RdbmsSource) source;
        String properties = ReflectUtil.fieldExists(RdbmsSource.class, "properties") ? RdbmsSource.getProperties() : "";
        return String.format(CP_POOL_KEY, RdbmsSource.getUrl(), RdbmsSource.getUsername(), RdbmsSource.getPassword(), properties);
    }

    public Connection setSchema(Connection conn, String schema) {
        if (StringUtils.isBlank(schema)) {
            return conn;
        }
        try {
            conn.setSchema(schema);
        } catch (Throwable e) {
            log.warn(String.format("setting schema exception : %s", e.getMessage()), e);
        }
        return conn;
    }
}
